package jrain.iot.server.mqtt.handler;

import java.util.HashMap;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import jrain.fw.core.business.BusinessConst;
import jrain.fw.core.business.BusinessRequest;
import jrain.fw.core.business.BusinessResponse;
import jrain.fw.core.business.client.BusinessInnerClient;
import jrain.fw.core.log.Logger;
import jrain.fw.core.log.LoggerFactory;
import jrain.fw.core.utils.FwUtils;
import jrain.iot.server.mqtt.business.MqttSimplePingService;
import jrain.iot.server.mqtt.consts.MqttConst;
import jrain.iot.server.mqtt.consts.MqttEventType;
import jrain.utils.collection.RDMap;
import jrain.utils.collection.RMap;
import jrain.utils.consts.BaseConst;
import jrain.utils.lang.LangUtils;
import jrain.utils.lang.StringUtils;

@Sharable
public class MqttHandler extends ChannelInboundHandlerAdapter {

	private static Logger logger = LoggerFactory.getLogger(MqttSimplePingService.class);
	private static final AttributeKey<String> sessionKey = AttributeKey.valueOf("sessionKey");
	private final BusinessInnerClient apiClient;
	private final String userServiceCode;
	private final String userServiceMethod;
	private final String eventServiceCode;
	private final String eventServiceMethod;
	private final String msgServiceCode;
	private final String msgServiceMethod;
	private final String pingServiceCode;
	private final String pingServiceMethod;
	private final String ctxServiceCode;
	private final String ctxServicePut;
	private final String ctxServiceRemove;

	public MqttHandler() {
		apiClient = FwUtils.getBusinessInnerClient();
		RMap<String, Object> cfgMap = FwUtils.getCfgService().getCfgMap("mqtt", new RDMap<>(new HashMap<>()));
		userServiceCode = StringUtils.trimNull(cfgMap.get("mqtt.user.service.code"), "MqttSimpleUserService");
		userServiceMethod = StringUtils.trimNull(cfgMap.get("mqtt.user.service.method"), "auth");
		eventServiceCode = StringUtils.trimNull(cfgMap.get("mqtt.event.service.code"), "MqttSimpleEventService");
		eventServiceMethod = StringUtils.trimNull(cfgMap.get("mqtt.event.service.method"), "handler");
		msgServiceCode = StringUtils.trimNull(cfgMap.get("mqtt.msg.service.code"), "MqttSimpleMsgService");
		msgServiceMethod = StringUtils.trimNull(cfgMap.get("mqtt.msg.service.method"), "publish");
		pingServiceCode = StringUtils.trimNull(cfgMap.get("mqtt.ping.service.code"), "MqttSimplePingService");
		pingServiceMethod = StringUtils.trimNull(cfgMap.get("mqtt.ping.service.method"), "ping");
		ctxServiceCode = StringUtils.trimNull(cfgMap.get("mqtt.ctx.service.code"), "MqttSimpleCtxService");
		ctxServicePut = StringUtils.trimNull(cfgMap.get("mqtt.ctx.service.put"), "put");
		ctxServiceRemove = StringUtils.trimNull(cfgMap.get("mqtt.ctx.service.publish"), "remove");
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		MqttMessage mqttMessage = (MqttMessage) msg;
		MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
		MqttMessageType mqttMessageType = mqttFixedHeader.messageType();
		if (mqttMessageType == MqttMessageType.CONNECT) {
			connect(ctx, (MqttConnectMessage) mqttMessage);
		} else if (mqttMessageType == MqttMessageType.PUBLISH) {
			if (isAuthorized(ctx)) {
				// 发送消息
				publish(ctx, (MqttPublishMessage) mqttMessage);
			}
		} else if (mqttMessageType == MqttMessageType.PINGREQ) {
			if (isAuthorized(ctx)) {
				// 心跳连接
				ping(ctx, mqttMessage);
			}
		} else if (mqttMessageType == MqttMessageType.DISCONNECT) {
			if (isAuthorized(ctx)) {
				// 断开连接
				disconnect(ctx, mqttMessage);
			}
		} else {
			logger.warn(BusinessConst.LOG_BOOT_REQUEST, "Message types are not supported messageType=={}", mqttMessageType);
			if (isAuthorized(ctx)) {
				connnectAckFuncNotFound(ctx);
			}
		}
		ReferenceCountUtil.release(msg);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		close(ctx, MqttEventType.DISCONN_500);
	}

	// 建立连接=========================================================================================
	public void connect(ChannelHandlerContext ctx, MqttConnectMessage connMessage) {
		Attribute<String> session = ctx.channel().attr(sessionKey);
		String userName = StringUtils.trimNull(connMessage.payload().userName());
		byte pwds[] = connMessage.payload().passwordInBytes();
		String password = "";
		if (pwds != null) {
			password = new String(pwds, BaseConst.UTF8);
		}
		// 登陆请求
		BusinessRequest request = new BusinessRequest();
		request.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
		request.setService(userServiceCode);
		request.setMethod(userServiceMethod);
		request.putData(MqttConst.USER_NAME, userName);
		request.putData(MqttConst.PASSWORD, password);
		request.putExtData(MqttConst.CTX, ctx);
		BusinessResponse response = apiClient.doSync(request);
		// 事件请求
		BusinessRequest eventRequest = new BusinessRequest();
		eventRequest.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
		eventRequest.setService(eventServiceCode);
		eventRequest.setMethod(eventServiceMethod);
		eventRequest.putData(MqttConst.USER_NAME, userName);
		eventRequest.putExtData(MqttConst.CTX, ctx);
		if (response.getCode() == BusinessConst.RES_CODE_SUCCESS) {
			session.set(userName);
			// 登录会话保存
			BusinessRequest ctxRequest = new BusinessRequest();
			ctxRequest.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
			ctxRequest.setService(ctxServiceCode);
			ctxRequest.setMethod(ctxServicePut);
			ctxRequest.putData("key", userName);
			ctxRequest.putData("value", ctx);
			apiClient.doSync(ctxRequest);
			MqttConnAckMessage ack = connnectAck(MqttConnectReturnCode.CONNECTION_ACCEPTED);
			ctx.writeAndFlush(ack);
			eventRequest.putData(MqttConst.EVENT_TYPE, MqttEventType.CONN_200.getCode());
			apiClient.doSync(eventRequest);

		} else {
			MqttConnAckMessage ack = connnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
			ctx.writeAndFlush(ack);
			eventRequest.putData(MqttConst.EVENT_TYPE, MqttEventType.CONN_403.getCode());
			apiClient.doSync(eventRequest);
		}
	}

	// 发布消息=========================================================================================
	public void publish(ChannelHandlerContext ctx, MqttPublishMessage publishMessage) {
		ByteBuf byteBuf = publishMessage.content();
		byte[] bytes = new byte[publishMessage.content().readableBytes()];
		byteBuf.readBytes(bytes);
		String userName = ctx.channel().attr(sessionKey).get();
		MqttQoS qoS = publishMessage.fixedHeader().qosLevel();
		int msgId = publishMessage.variableHeader().packetId();
		String topicName = publishMessage.variableHeader().topicName();
		if (qoS == MqttQoS.AT_LEAST_ONCE) {
			// 登陆请求
			BusinessRequest request = new BusinessRequest();
			request.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
			request.setService(msgServiceCode);
			request.setMethod(msgServiceMethod);
			request.putData(MqttConst.USER_NAME, userName);
			request.putData(MqttConst.MSG_DATA, bytes);
			request.putData(MqttConst.TOPIC_NAME, topicName);
			request.putExtData(MqttConst.CTX, ctx);
			BusinessResponse response = apiClient.doSync(request);
			if (response.getCode() == BusinessConst.RES_CODE_SUCCESS) {
				MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.FAILURE, false, 0x02);
				MqttMessageIdVariableHeader messageIdVariableHeader = MqttMessageIdVariableHeader.from(msgId);
				MqttPubAckMessage ackMessage = new MqttPubAckMessage(mqttFixedHeader, messageIdVariableHeader);
				ctx.writeAndFlush(ackMessage);
			} else {
				connnectAckServiceUnavaliable(ctx);
			}
		} else if (qoS == MqttQoS.AT_MOST_ONCE) {
			BusinessRequest request = new BusinessRequest();
			request.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
			request.setService(msgServiceCode);
			request.setMethod(msgServiceMethod);
			request.putExtData(MqttConst.CTX, ctx);
			request.putData(MqttConst.TOPIC_NAME, topicName);
			request.putData(MqttConst.USER_NAME, userName);
			request.putData(MqttConst.MSG_DATA, bytes);
			apiClient.doSync(request);
		}
	}

	// 心跳连接=========================================================================================
	public void ping(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
		String userName = ctx.channel().attr(sessionKey).get();
		BusinessRequest eventRequest = new BusinessRequest();
		eventRequest.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
		eventRequest.setService(pingServiceCode);
		eventRequest.setMethod(pingServiceMethod);
		eventRequest.putData(MqttConst.USER_NAME, userName);
		eventRequest.putExtData(MqttConst.CTX, ctx);
		apiClient.doSync(eventRequest);
		byte[] pingresp = { (byte) 0xd0, 0x00 };
		ByteBuf buff = ctx.alloc().buffer();
		buff.writeBytes(pingresp);
		ctx.writeAndFlush(buff);
	}

	// 断开连接=========================================================================================
	public void disconnect(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
		// 返回断开连接消息
		byte[] message = { (byte) 0xe0, 0x00 };
		ByteBuf buff1 = ctx.alloc().buffer();
		buff1.writeBytes(message);
		ctx.writeAndFlush(buff1);
		close(ctx, MqttEventType.DISCONN_200);
	}
	// 关闭连接=========================================================================================

	public void close(ChannelHandlerContext ctx, MqttEventType eventType) {
		String userName = getUserName(ctx);
		close(ctx, eventType, userName);
	}

	public void close(ChannelHandlerContext ctx, MqttEventType eventType, String userName) {
		// 通知客户端下线
		ctx.close();
		if (!StringUtils.isEmpty(userName)) {
			// 事件请求
			BusinessRequest eventRequest = new BusinessRequest();
			eventRequest.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
			eventRequest.setService(eventServiceCode);
			eventRequest.setMethod(eventServiceMethod);
			eventRequest.putData(MqttConst.USER_NAME, userName);
			eventRequest.putData(MqttConst.EVENT_TYPE, eventType.getCode());
			apiClient.doSync(eventRequest);
			// 退出
			BusinessRequest ctxRequest = new BusinessRequest();
			ctxRequest.putData(BusinessConst.EXT_TRACE_ID, LangUtils.randomUUID());
			ctxRequest.setService(ctxServiceCode);
			ctxRequest.setMethod(ctxServiceRemove);
			ctxRequest.putData("key", userName);
			ctxRequest.putData("value", ctx);
			apiClient.doSync(ctxRequest);
		}

	}

	// 权限认证=========================================================================================
	public String getUserName(ChannelHandlerContext ctx) {
		Attribute<String> session = ctx.channel().attr(sessionKey);
		return session != null ? session.get() : null;
	}

	public boolean isAuthorized(ChannelHandlerContext ctx) {
		String userName = getUserName(ctx);
		if (StringUtils.isEmpty(userName)) {
			connnectAckNotAuthorized(ctx);
			return false;
		}
		return true;
	}

	// 连接响应=========================================================================================
	public void connnectAckNotAuthorized(ChannelHandlerContext ctx) {
		MqttConnAckMessage ack = connnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
		ctx.writeAndFlush(ack);
		close(ctx, MqttEventType.DISCONN_403);
	}

	public void connnectAckServiceUnavaliable(ChannelHandlerContext ctx) {
		MqttConnAckMessage ack = connnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
		ctx.writeAndFlush(ack);
		close(ctx, MqttEventType.DISCONN_500);
	}

	public void connnectAckFuncNotFound(ChannelHandlerContext ctx) {
		MqttConnAckMessage ack = connnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
		ctx.writeAndFlush(ack);
		close(ctx, MqttEventType.DISCONN_404);
	}

	public MqttConnAckMessage connnectAck(MqttConnectReturnCode returnCode) {
		MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, false);
		MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.FAILURE, false, 0x02);
		MqttConnAckMessage ackMessage = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
		return ackMessage;
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state().equals(IdleState.ALL_IDLE)) {
				close(ctx, MqttEventType.DISCONN_999);
			} else {
				super.userEventTriggered(ctx, evt);
			}
		} else {
			super.userEventTriggered(ctx, evt);
		}
	}
}
